package com.zhang.gmall.utils;

import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;


/**
 * @title:
 * @author: zhang
 * @date: 2022/3/23 20:52
 */
public class MyEnv {
    private static StreamExecutionEnvironment env;

    public static StreamExecutionEnvironment getStreamingEnv(int parallelism, boolean isCheckPoint) {
        env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(parallelism);
        if (isCheckPoint) {
            env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
            env.setStateBackend(new FsStateBackend("hdfs://hadoop102:8020/flink/gamll-dev"));
            env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L);
            env.getCheckpointConfig().setMinPauseBetweenCheckpoints(2000L);
            env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
            env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 1000L));
            System.setProperty("HADOOP_USER_NAME", "zhang");
        }
        return env;
    }




}
